use std::{
    collections::HashMap,
    sync::{Arc, RwLock},
};

use crate::transport::{get_sender, is_setuped, remove_connection};

use super::message::Message;
use rand::Rng;
use tokio::{
    sync::mpsc::UnboundedSender,
    time::{sleep, timeout, Duration},
};

lazy_static::lazy_static! {
    //
    pub(crate) static ref RESPONSE_BUFFER: RwLock<HashMap<u64, Option<Message>>> = RwLock::new(HashMap::new());
}

pub async fn request_response(
    peer_id: &String,
    message: Message,
    time_out: u64,
) -> anyhow::Result<Message> {
    //获取sender
    let sender = get_sender(peer_id)?;

    tracing::debug!("开始请求.....",);
    if let Ok(response_result) = timeout(
        Duration::from_millis(time_out),
        Box::pin(async move {
            //发送结果
            _response(sender, message).await
        }),
    )
    .await
    {
        let response = response_result.map_err(|err| anyhow::anyhow!("{}", err))?;
        tracing::debug!("返回数据成功!");
        return Ok(response);
    } else {
        tracing::debug!("请求超时!");
        return Ok(Message::timeout());
    }
}
pub async fn fire_and_forget(peer_id: &String, message: Message) -> anyhow::Result<()> {
    //获取sender
    let sender = get_sender(peer_id)?;

    if let Err(err) = sender.send(message) {
        tracing::error!("发送失败:{}", err);
        return Err(anyhow::anyhow!("发送失败:{}", err));
    }
    Ok(())
}

async fn _response(
    sender: UnboundedSender<Message>,
    mut message: Message,
) -> anyhow::Result<Message> {
    //设置一个
    let request_id = generate_push_request_id()?;
    tracing::debug!("请求ID:{}", request_id);
    message.request_id = request_id;

    let cmd = message.command.clone();

    if let Err(err) = sender.send(message) {
        //发送失败
        let _ = remove_response(request_id).await;

        return Ok(Message::error_with_code_message(
            1,
            format!("{}", err).as_str(),
            cmd,
        ));
    }
    //发送成功,循环进行等待结果
    loop {
        let pop_res = pop_response(request_id).await;
        match pop_res {
            Ok(res_message) => {
                if let Some(response) = res_message {
                    //获取到返回值
                    return Ok(response);
                } else {
                    //
                    sleep(Duration::from_millis(10)).await;
                }
            }
            Err(err) => {
                let _ = remove_response(request_id).await;

                return Ok(Message::error_with_code_message(
                    2,
                    format!("{}", err).as_str(),
                    cmd,
                ));
            }
        }
    }
}

pub fn generate_push_request_id() -> anyhow::Result<u64> {
    //随机一个值
    let mut rng = rand::thread_rng();

    let mut request_id: u64 = rng.gen();
    while !push_response(request_id)? {
        request_id = rng.gen();
    }
    Ok(request_id)
}

pub fn push_response(request_id: u64) -> anyhow::Result<bool> {
    //判断是否存在
    let exist = {
        let cache = RESPONSE_BUFFER
            .read()
            .map_err(|err| anyhow::anyhow!("{}", err))?;
        cache.contains_key(&request_id)
    };
    if exist {
        Ok(false)
    } else {
        //
        let mut cache_result = RESPONSE_BUFFER
            .write()
            .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;

        cache_result.insert(request_id, None);
        tracing::debug!("插入成功!");
        Ok(true)
    }
}

pub async fn pop_response(request_id: u64) -> anyhow::Result<Option<Message>> {
    //
    let mut cache_result = RESPONSE_BUFFER
        .write()
        .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;
    if let Some(message) = cache_result.get(&request_id) {
        if message.is_some() {
            if let Some(message) = cache_result.remove(&request_id) {
                Ok(message)
            } else {
                Err(anyhow::anyhow!("不存在该请求id!{}", request_id))
            }
        } else {
            Ok(None)
        }
    } else {
        Err(anyhow::anyhow!("不存在该请求id!{}", request_id))
    }
}

pub async fn remove_response(request_id: u64) -> anyhow::Result<Option<Message>> {
    //
    let mut cache_result = RESPONSE_BUFFER
        .write()
        .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;
    if let Some(message) = cache_result.remove(&request_id) {
        Ok(message)
    } else {
        Ok(None)
    }
}

pub async fn put_response(response: Message) -> anyhow::Result<()> {
    let request_id = response.request_id;
    //
    let mut cache_result = RESPONSE_BUFFER
        .write()
        .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;
    if let Some(message) = cache_result.get_mut(&request_id) {
        *message = Some(response);
        Ok(())
    } else {
        Err(anyhow::anyhow!("不存在该请求id!{}", request_id))
    }
}

pub async fn ping(peer_id: &String) -> anyhow::Result<()> {
    if let Ok(setuped) = is_setuped(peer_id) {
        if setuped {
            if let Err(err) = fire_and_forget(&peer_id, Message::ping()).await {
                tracing::debug!("Ping发送失败,关闭连接!{}", err);
                return remove_connection(peer_id);
            }
        }
    }
    Ok(())
}
